package org.example.realtime.jtp.common.utils;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

/**
 * 将DataStream流式数据实时保存MySQL数据库、ClickHouse数据库，采用内置JdbcSink实现。
 * @author xuanyu
 */
public class JdbcUtil {

	/**
	 * 将DataStream数据流中数据保存到MySQL表，各个字段之间值使用逗号隔开
	 * @param stream 数据流，数据类型为String
	 * @param upsertSql 插入数据库语句
	 */
	public static void sinkToMysqlUpsert(DataStream<String> stream, String upsertSql){
		// a. sink对象
		SinkFunction<String> sink = JdbcSink.<String>sink(
			upsertSql,
			(statement, value) -> {
				String[] split = value.split(",");
				for(int i = 0; i < split.length; i ++){
					statement.setObject(i + 1, split[i]);
				}
			},
			JdbcExecutionOptions.builder()
				.withBatchSize(1000)
				.withBatchIntervalMs(200)
				.withMaxRetries(5)
				.build(),
			new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
				.withUrl("jdbc:mysql://node101:3306/?useSSL=false&useUnicode=true&characterEncoding=UTF-8")
				.withDriverName("com.mysql.cj.jdbc.Driver")
				.withUsername("root")
				.withPassword("123456")
				.build()
		);
		// b. 保存数据
		stream.addSink(sink);
	}

	/**
	 * 将DataStream数据流中数据保存到Clickhouse表，各个字段之间值使用逗号隔开
	 * @param stream 数据流，数据类型为String
	 * @param upsertSql 插入数据库语句
	 */
	public static void sinkToClickhouseUpsert(DataStream<String> stream, String upsertSql){
		// a. sink对象
		SinkFunction<String> sink = JdbcSink.<String>sink(
			upsertSql,
			(statement, value) -> {
				String[] split = value.split(",");
				for(int i = 0; i < split.length; i ++){
					statement.setObject(i + 1, split[i]);
				}
			},
			JdbcExecutionOptions.builder()
				.withBatchSize(1000)
				.withBatchIntervalMs(200)
				.withMaxRetries(5)
				.build(),
			new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
				.withUrl("jdbc:clickhouse://node103:8123")
				.withDriverName("ru.yandex.clickhouse.ClickHouseDriver")
				.withUsername("default")
				.withPassword("")
				.build()
		);
		// b. 保存数据
		stream.addSink(sink);
	}

	/**
	 * 将DataStream数据流中数据保存到Clickhouse表，每条数据为JSON字符串
	 * @param stream 数据流，数据类型为String
	 * @param upsertSql 插入数据库语句
	 */
	public static void sinkToClickhouseUpsertJson(DataStream<String> stream, String[] columns, String upsertSql){
		// a. sink对象
		SinkFunction<String> sink = JdbcSink.<String>sink(
			upsertSql,
			(statement, value) -> {
				// 解析json字符串
				JSONObject jsonObject = JSON.parseObject(value);
				for(int i = 0; i < columns.length; i ++){
					// 列名称
					String columnName = columns[i];
					// 列值
					Object columnValue = jsonObject.get(columnName);
					// 设置占位符值
					statement.setObject(
						i + 1,
						columnValue.toString()
					);
				}
			},
			JdbcExecutionOptions.builder()
				.withBatchSize(1000)
				.withBatchIntervalMs(200)
				.withMaxRetries(5)
				.build(),
			new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
				.withUrl("jdbc:clickhouse://node103:8123")
				.withDriverName("ru.yandex.clickhouse.ClickHouseDriver")
				.withUsername("default")
				.withPassword("")
				.build()
		);
		// b. 保存数据
		stream.addSink(sink);
	}

}
